package com.xiguanlezz.cn.consumer.future;

import com.xiguanlezz.cn.FutureConst;
import com.xiguanlezz.cn.consumer.callback.AsyncRpcCallback;
import com.xiguanlezz.cn.consumer.callback.ClientThreadPool;
import com.xiguanlezz.cn.protocol.RpcHeader;
import com.xiguanlezz.cn.protocol.RpcProtocol;
import com.xiguanlezz.cn.protocol.RpcRequest;
import com.xiguanlezz.cn.protocol.RpcResponse;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/5 19:09
 * @Description：基于Future和自定义AQS的异步转同步实现类
 */
public class RpcFuture implements Future<Object> {
    private RpcProtocol<RpcRequest> request;
    private RpcProtocol<RpcResponse> response;
    private long startTime;

    private ReentrantLock lock = new ReentrantLock();
    private List<AsyncRpcCallback> pendingCallbacks = new ArrayList<>();

    // 自定义同步器。相当于就是一对一的线程间通信，也可以借助CountDownLatch完成
    private Sync sync = new Sync();

    public static RpcFuture createRpcFuture(RpcProtocol<RpcRequest> request) {
        RpcFuture rpcFuture = new RpcFuture();
        rpcFuture.request = request;
        rpcFuture.startTime = System.currentTimeMillis();
        return rpcFuture;
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        return Objects.nonNull(response);
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        // 获取多少资源即方法参数是啥不重要，毕竟自定义了AQS
        sync.acquire(1);

        if (isDone()) {
            return Optional.ofNullable(response)
                    .map(RpcProtocol::getBody)
                    .map(RpcResponse::getResult)
                    .orElse(null);
        }
        return null;
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        // 带超时时间的线程阻塞
        if (sync.tryAcquireNanos(1, unit.toNanos(timeout))) {
            if (isDone()) {
                return Optional.ofNullable(response)
                        .map(RpcProtocol::getBody)
                        .map(RpcResponse::getResult)
                        .orElse(null);
            }
        } else {
            RpcHeader rpcHeader = request.getHeader();
            RpcRequest body = request.getBody();
            throw new RuntimeException(
                    String.format(
                            FutureConst.FUTURE_TIMEOUT_FORMAT,
                            rpcHeader.getMessageId(),
                            body.getClassName(),
                            body.getMethodName()
                    )
            );
        }
        return null;
    }

    public void done(RpcProtocol<RpcResponse> response) {
        this.response = response;
        // 释放多少资源即方法参数是啥不重要，毕竟自定义了AQS
        sync.release(1);

        // 调用回调方法
        if (request.getBody().isNeedCallback()) {
            invokeCallbacks();
        }
    }

    public void addCallback(AsyncRpcCallback callback) {
        lock.lock();
        try {
            if (isDone()) {
                // 如果任务已经完成则直接执行回调
                runCallback(callback);
            } else {
                // 如果任务还未完成，则放入pending列表中
                pendingCallbacks.add(callback);
            }
        } finally {
            lock.unlock();
        }
    }

    private void invokeCallbacks() {
        lock.lock();
        try {
            for (AsyncRpcCallback callback : pendingCallbacks) {
                runCallback(callback);
            }
        } finally {
            lock.unlock();
        }
    }

    private void runCallback(AsyncRpcCallback callback) {
        RpcResponse body = response.getBody();
        ClientThreadPool.submit(() -> {
            if (Objects.isNull(body.getError())) {
                callback.onSuccess(body.getResult());
            } else {
                callback.onError(body.getError());
            }
        });
    }

    private static class Sync extends AbstractQueuedSynchronizer {
        static final int noResult = 0;
        static final int hasResult = 1;

        @Override
        protected boolean tryAcquire(int arg) {
            return compareAndSetState(hasResult, noResult);
        }

        @Override
        protected boolean tryRelease(int arg) {
            return compareAndSetState(noResult, hasResult);
        }
    }

}
